
#include "linux_def.h"
#include "channel.h"
#include "connection.h"
#include "event.h"
#include "cycle.h"
#include "process.h"

static void start_worker_processes(cycle_t *cycle, int n);
static void signal_worker_processes(cycle_t *cycle, int signo);
static int reap_children(cycle_t *cycel);
static void master_process_exit(cycle_t *cycle);
static void worker_process_cycle(cycle_t *cycle, void *data);
static void worker_process_init(cycle_t *cycle);
static void worker_process_exit(cycle_t *cycle);
static void channel_handler(event_t *ev);

#if (THREADS)
static void wakeup_worker_threads(cycle_t *cycle);
static void *worker_thread_cycle(void *data);
#endif

int	process;
pid_t	pid;
int	threaded;

sig_atomic_t	reap;
sig_atomic_t	sigio;
sig_atomic_t	sigalrm;
sig_atomic_t	terminate;
sig_atomic_t	quit;

int	exiting;

#if (THREADS)
#endif

static int reap_children(cycle_t *cycle)
{
	int	i, n;
	int	live;
	channel_t	ch;

	ch.command = CMD_CLOSE_CHANNEL;
	ch.fd = -1;

	live = 0;
	for (i=0; i<process_last; i++){
		if (processes[i].pid == -1)
			continue;
		
		if (processes[i].exited){
			if (!processes[i].detached){
				close_channel(processes[i].channel);
				processes[i].channel[0] = -1;
				processes[i].channel[1] = -1;

				ch.pid = processes[i].pid;
				ch.slot = i;

				for (n=0; n<process_last; n++){
					if (processes[n].exited || processes[n].pid == -1 || processes[n].channel[0] == -1)
						continue;

					write_channel(processes[i].channel[0], &ch, sizeof(channel_t));
				}
			}

			if (processes[i].respawn && !processes[i].exiting && !terminate && !quit){
				if (spawn_process(cycle, processes[i].proc, processes[i].data, processes[i].name, i) == E_FAIL)
					continue;
	
				ch.command = CMD_OPEN_CHANNEL;
				ch.pid = processes[process_slot].pid;
				ch.slot = process_slot;
				ch.fd = processes[process_slot].channel[0];
	
				for (n=0; n<process_last; n++){
					if (n == process_slot || processes[n].pid==-1 || processes[n].channel[0]==-1)
						continue;
					write_channel(processes[i].channel[0], &ch, sizeof(channel_t));
				}
				
				live = 1;
				continue;
			}

			if (i == process_last-1)
				process_last--;
			else
				processes[i].pid = -1;
		}else if (processes[i].exiting || !processes[i].detached){
			live = 1;
		}
	}

	return live;
}

static void master_process_exit(cycle_t *cycle)
{
	destroy_pool(cycle->pool);
	
	exit(0);
}


static void signal_worker_processes(cycle_t *cycle, int signo)
{
	int	i;
	channel_t	ch;
	
	switch(signo){
		case SIGQUIT:
			ch.command = CMD_QUIT;
			break;
		case SIGTERM:
			ch.command = CMD_TERMINATE;
			break;
		default:
			ch.command = 0;
	}

	ch.fd = -1;

	for (i=0; i<process_last; i++){
		if (processes[i].detached || processes[i].pid == -1)
			continue;
	
		if (processes[i].just_spawn){
			processes[i].just_spawn = 0;
			continue;
		}
		
		if (processes[i].exiting && signo == SIGQUIT)
			continue;

		if (ch.command){
			if(write_channel(processes[i].channel[0], &ch, sizeof(channel_t)) == E_OK){
				processes[i].exiting = 1;
				continue;
			}
		}

		if (kill(processes[i].pid, signo) == E_FAIL){
			if (errno == ESRCH){	//进程不存在
				processes[i].exited = 1;
				processes[i].exiting = 0;
				reap = 1;
			}
			continue;
		}
		processes[i].exiting = 1;
	}
}


void master_process_cycle(cycle_t *cycle)
{
	int	n;
	sigset_t	set;
	listening_t	*ls;
	struct itimerval	itv;

	sigemptyset(&set);	
	sigaddset(&set, SIGCHLD);
	sigaddset(&set, SIGALRM);
	sigaddset(&set, SIGIO);
	sigaddset(&set, SIGHUP);
	sigaddset(&set, SIGUSR1);
	sigaddset(&set, SIGUSR2);
	sigaddset(&set, SIGINT);
	sigaddset(&set, SIGTERM);
	sigaddset(&set, SIGQUIT);
	
	if (sigprocmask(SIG_BLOCK, &set, NULL) == -1){
		fprintf(stderr, "sigpromask faile.\n");
		return;
	}

	sigemptyset(&set);	

	start_worker_processes(cycle, 3);

	int	delay	= 0;
	int	live	= 1;

	for (;;){
		if (delay){
			delay	*= 2;

			itv.it_interval.tv_sec	= 0;
			itv.it_interval.tv_usec = 0;
			itv.it_value.tv_sec	= delay / 1000;
			itv.it_value.tv_usec = (delay % 1000) * 1000;
		
			if (setitimer(ITIMER_REAL, &itv, NULL) == -1){
				fprintf(stderr, "setitimer() failed.\n");
			}
		}

		sigsuspend(&set);

		time_update(0, 0);

		if (reap){
			reap	= 0;
			live	= reap_children(cycle);
		}
	
		if (!live && (terminate || quit)){
			if (verbose)
				fprintf(stderr, "== %d == In master_process_cycle() terminate.\n", __LINE__);
			master_process_exit(cycle);
		}

		if (terminate){
			printf("Master recv SIGTERMINATE\n");
			if (delay == 0)
				delay = 50;

			if (delay > 1000){
				signal_worker_processes(cycle, SIGKILL);
			}else{
				signal_worker_processes(cycle, SIGTERM);
			}
			continue;
		}

		if (quit){
			printf("Master recv SIGQUIT\n");
			signal_worker_processes(cycle, SIGQUIT);

			ls	= cycle->listening.elts;
			for(n=0; n<cycle->listening.nelts; n++){
				close(ls[n].fd);
			}
			cycle->listening.nelts = 0;
			continue;
		}
	}
}

void start_worker_processes(cycle_t *cycle, int n)
{
	int	i, s;
	channel_t	ch;

	ch.command = CMD_OPEN_CHANNEL;

	for (i=0; i<n; i++){
		spawn_process(cycle, worker_process_cycle, NULL, "worker process", PROCESS_RESPAWN);
		
		ch.pid	= processes[process_slot].pid;
		ch.slot	= process_slot;
		ch.fd	= processes[process_slot].channel[0];
		
		for (s=0; s<process_last; s++){
			if (s == process_slot
				|| processes[s].pid == -1
				|| processes[s].channel[0] == -1)
			{
				continue;
			}

			write_channel(processes[s].channel[0], &ch, sizeof(channel_t));
		}
		sleep(1);
	}
}

static void worker_process_init(cycle_t *cycle)
{
	sigset_t	set;
	int	n, i;
	listening_t	*ls;

	process = PROCESS_WORKER;

	sigemptyset(&set);
	if (sigprocmask(SIG_SETMASK, &set, NULL) == -1){
		fprintf(stderr, "sigpromask() failed.\n");
	}

	//事件的初始化
	event_process_init(cycle);	

	//关闭不许要的文件描述符
	for (n=0; n<process_last; n++){
		if (processes[n].pid == -1)
			continue;
		if (n == process_slot)
			continue;
		if (processes[n].channel[1] == -1)
			continue;

		if (close(processes[n].channel[1]) == -1)
			fprintf(stderr, "close(channel) failed.\n");
	}

	if (close(processes[process_slot].channel[0]) == -1)
		fprintf(stderr, "close(channel) failed.\n");


	add_channel_event(cycle, g_channel, READ_EVENT, channel_handler);
}

static void channel_handler(event_t *ev)
{
	int	n;
	channel_t	ch;
	connection_t	*c;

	if (ev->timedout){
		ev->timedout = 0;
		return;
	}

	c = ev->data;

	for (;;){

		n = read_channel(c->fd, &ch, sizeof(channel_t));
		
		if (n == -1){
			if (event_flags & USE_EPOLL_EVENT)
				epoll_del_connection(c, 0);

			close_connection(c);
			return;
		}

		if (n == EAGAIN)
			return;

		switch(ch.command){
			case CMD_QUIT:
				quit	= 1;
				break;
			case CMD_TERMINATE:
				terminate = 1;
				break;
			case CMD_OPEN_CHANNEL:
				processes[ch.slot].pid = ch.pid;
				processes[ch.slot].channel[0] = ch.fd;
				break;
			case CMD_CLOSE_CHANNEL:
				close(processes[ch.slot].channel[0]);
				processes[ch.slot].channel[0] = -1;
				break;
			default:
				break;
		}
	}
}

void worker_process_cycle(cycle_t *cycle, void *data)
{
	int i;
	connection_t	*c;

	worker_process_init(cycle);

	for (;;){
	/*	if (exiting){
			c = cycle->connections;			
			for (i=0; i<cycle->connection_n; i++){
				if (c[i].fd != -1 && c[i].idle)	{
					c[i].close = 1;
					c[i].read->handler(c[i].read);
				}
			}

			if (event_timer_rbtree.root == event_timer_rbtree.sentinel)
				worker_process_exit(cycle);
		}
	*/	
		process_events_and_timers(cycle);

		if (quit){
			quit = 0;
			if (verbose)
				fprintf(stderr, "quit-- %d -- child quit\n", getpid());
			if (!exiting){
				close_listening_sockets(cycle);
				exiting  = 1;
			}
		}

		if (terminate){
			if (verbose)
				fprintf(stderr, "terminate-- %d -- child terminate\n", getpid());
			worker_process_exit(cycle);
		}

	}
}

static void worker_process_exit(cycle_t *cycle)
{
	int	i;
	connection_t	*c;

	if (exiting){
		c = cycle->connections;
		for (i=0; i<cycle->connection_n; i++){
			if (c[i].fd!=-1 && c[i].read && !c[i].read->accept){
				raise(SIGSTOP);	
			}
		}

		raise(SIGSTOP);	
	}
	

	destroy_pool(cycle->pool);

	fprintf(stderr, "-- %d -- child process exit.\n", getpid());

	exit(0);
}
